package com.beiding.dbrequest.client;

import com.beiding.dbrequest.common.*;
import com.beiding.dbrequest.config.RetryConfig;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.timeout.IdleStateHandler;
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.java.Log;

import java.io.Serializable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 客户端封装
 *
 * @author 丁常磊
 * @date 2021/7/15 16:18
 */
@Log
public class Client extends Terminal {

    @Setter
    @Getter
    private String targetHost;

    @Setter
    @Getter
    private Integer targetPort;

    @Getter
    private ChannelHandlerContext channelHandlerContext;

    @Getter
    private SocketChannel socketChannel;

    @Getter
    private NioEventLoopGroup workerGroup;

    @Setter
    @Getter
    private RetryConfig retryConfig;

    /**
     * 执行线程数
     */
    @Setter
    @Getter
    private int workerThreadNumber = 1;

    /**
     * 如果指定workerThreadNumber为负数则使用默认worker
     */
    private static NioEventLoopGroup DEFAULT_WORKER;

    private static NioEventLoopGroup getDefaultWorker() {
        if (DEFAULT_WORKER != null) {
            return DEFAULT_WORKER;
        }
        synchronized (Client.class) {
            if (DEFAULT_WORKER == null) {
                DEFAULT_WORKER = new NioEventLoopGroup();
            }
        }
        return DEFAULT_WORKER;
    }


    public Observable<?> start() {
        return Observable.create(subscriber -> {
            //客户端不可以被重复启动
            synchronized (Client.this) {
                if (workerGroup != null) {
                    subscriber.onError(new IllegalStateException("客户端不可重复启动"));
                    return;
                }
                //启动netty服务
                if (workerThreadNumber < 0) {
                    workerGroup = getDefaultWorker();
                } else {
                    workerGroup = new NioEventLoopGroup(workerThreadNumber);
                }

                //传递事件
                connect().subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Object o) {

                    }

                    @Override
                    public void onError(Throwable e) {

                        if (retryConfig != null) {
                            if (retryConfig.isRetryOnStart()) {
                                retry();
                            } else {
                                subscriber.onError(e);
                            }
                        } else {
                            subscriber.onError(e);
                        }

                    }

                    @Override
                    public void onComplete() {
                        subscriber.onComplete();
                    }
                });
            }
        });
    }


    /**
     * 关闭客户端
     */
    public Observable<?> close() {

        return Observable.create(subscriber -> {

            // 任务数 所有的任务数减到0表示全部完成
            AtomicInteger atomicInteger = new AtomicInteger();

            if (socketChannel != null) {
                atomicInteger.addAndGet(1);
            }

            if (workerGroup != null && workerGroup != DEFAULT_WORKER) {
                atomicInteger.addAndGet(1);
            }


            //关闭通道
            if (socketChannel != null) {
                socketChannel.closeFuture().addListener(future -> {
                    if (future.isSuccess()) {
                        channelHandlerContext = null;
                        socketChannel = null;
                        if (atomicInteger.addAndGet(-1) == 0) {
                            subscriber.onComplete();
                        }
                    } else {
                        subscriber.onError(future.cause());
                    }
                });

            }

            //关闭线程组
            if (workerGroup != null && workerGroup != DEFAULT_WORKER) {
                workerGroup.shutdownGracefully().addListener(future -> {
                    if (future.isSuccess()) {
                        workerGroup = null;
                        if (atomicInteger.addAndGet(-1) == 0) {
                            subscriber.onComplete();
                        }
                    } else {
                        subscriber.onError(future.cause());
                    }
                });
            }

            if (atomicInteger.addAndGet(0) == 0) {
                subscriber.onComplete();
            }

        });
    }


    /**
     * 创建连接
     *
     * @return
     */
    private Observable<?> connect() {

        return Observable.create(subscriber -> {

            try {
                //尝试创建连接
                Bootstrap bootstrap = new Bootstrap();


                //创建连接对象
                bootstrap.group(workerGroup)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) {
                                ch.pipeline()
                                        .addLast(new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS))
                                        .addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, lengthFieldLength, 0, lengthFieldLength))
                                        .addLast(new LengthFieldPrepender(lengthFieldLength))
                                        .addLast(new HessianEncoder())
                                        .addLast(new HessianDecoder())
                                        .addLast(new H() {
                                            @Override
                                            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                                super.channelActive(ctx);
                                                channelHandlerContext = ctx;
                                                socketChannel = (SocketChannel) ctx.channel();

                                                //创建连接完成
                                                subscriber.onComplete();
                                            }

                                            @Override
                                            public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                                                super.channelInactive(ctx);
                                                //通道被关闭的时候
                                                onChannelClose();
                                            }

                                            @Override
                                            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                                                onError(cause);
                                            }
                                        })
                                ;

                            }
                        });

                bootstrap.connect(targetHost, targetPort).addListener(future -> {
                    if (!future.isSuccess()) {
                        //连接失败时返回错误
                        subscriber.onError(future.cause());
                    }
                });
            } catch (Throwable throwable) {
                subscriber.onError(throwable);
            }

        });

    }

    /**
     * 重试延时
     */
    private long retryDelay = -1;

    /**
     * 重试次数
     */
    private long retryTime = 0;

    /**
     * 通道被关闭的时候
     */
    public void onChannelClose() {

        retry();

    }

    private void retry() {
        //重新尝试建立连接
        if (retryConfig != null) {
            if (retryConfig.getMaxRetryTime() > 0 && retryTime > retryConfig.getMaxRetryTime()) {
                //到最大重试次数 关闭 不再重试
                log.warning("超出最大重试次数,不再重试");
                close();
                return;
            }

            retryTime++;

            if (retryDelay == -1) {
                retryDelay = retryConfig.getFirstDelay();
            } else {
                retryDelay += retryConfig.getDelayStep();
            }
            if (retryDelay > retryConfig.getMaxDelay()) {
                retryDelay = retryConfig.getMaxDelay();
            }


            log.info("重试 延时[" + retryDelay + "] 次数[" + retryTime + "]");

            //延时重试
            SecondWorker.worker.schedule(() -> {

                //重新连接
                connect().subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Object o) {

                    }

                    @Override
                    public void onError(Throwable e) {
                        //再次重试
                        retry();
                    }

                    @Override
                    public void onComplete() {
                        //复位
                        retryDelay = -1;
                        retryTime = 0;

                        if (retryConfig != null) {
                            Runnable afterRetrySuccess = retryConfig.getAfterRetrySuccess();
                            try {
                                afterRetrySuccess.run();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }

                    }
                });

            }, retryDelay, TimeUnit.MILLISECONDS);

        }
    }

    /**
     * 当有异常情况时
     *
     * @param cause
     */
    public void onError(Throwable cause) {

    }

    /**
     * 发起请求
     *
     * @param host
     * @param port
     * @param request
     * @param responseCallback
     */
    public static void request(String host, int port, Message request, ResponseCallback responseCallback) {

        Client client = new Client();
        client.setTargetHost(host);
        client.setTargetPort(port);
        //指定单个线程
        client.setWorkerThreadNumber(-1);
        client.start()
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Object o) {

                    }

                    @Override
                    public void onError(Throwable e) {
                        Message resp = Message.response(Message.STATUS_INVOKE_ERROR, "无法连接服务端", null);
                        responseCallback.onResponse(resp);
                        //调用一次关闭 避免某些错误的资源创建
                        client.close().subscribe();
                    }

                    @Override
                    public void onComplete() {
                        client.request(request, new ResponseCallback() {
                            @Override
                            public void onResponse(Message responseMassage) {
                                responseCallback.onResponse(responseMassage);
                                //关闭
                                client.close().subscribe();
                            }
                        });
                    }
                });

    }

    /**
     * 发起请求不注册
     *
     * @param host
     * @param port
     * @param request
     * @return
     */
    public static Observable<Serializable> request(String host, int port, Message request) {
        return Observable.create(subscriber -> {
            request(host, port, request, new ObservableResponseCallback(subscriber));
        });
    }


    /**
     * 一次性请求头
     *
     * @param host
     * @param port
     * @param uri
     * @param body
     * @return
     */
    public static Observable<Serializable> request(String host, int port, String uri, Serializable body) {
        return Observable.create(subscriber -> {
            request(host, port, Message.request(uri, body), new ObservableResponseCallback(subscriber));
        });
    }


    /**
     * 使用内部通道发起请求头
     *
     * @param message
     * @param responseCallback
     */
    public void request(Message message, ResponseCallback responseCallback) {
        //严格幂等
        try {
            if (channelHandlerContext == null) {
                if (responseCallback != null) {
                    Message resp = Message.response(Message.STATUS_INVOKE_ERROR, "尚未连接成功", null);
                    responseCallback.onResponse(resp);
                }
            } else {
                request(channelHandlerContext, message, responseCallback);
            }
        } catch (Exception e) {
            responseCallback.onResponse(Message.response(Message.STATUS_INVOKE_ERROR, e.getMessage(), null));
        }
    }


    /**
     * 发起请求返回promise
     *
     * @param message
     * @return
     */
    public Observable<Serializable> request(Message message) {
        return Observable.create(subscriber -> {
            request(message, new ObservableResponseCallback(subscriber));
        });
    }

    /**
     * 发起请求
     *
     * @param uri
     * @param body
     * @return
     */
    public Observable<Serializable> request(String uri, Serializable body) {
        return request(Message.request(uri, body));
    }


    public Observable<?> broadcast(Message message) {
        if (channelHandlerContext == null) {
            return Observable.error(new IllegalStateException("通道不存在"));
        }
        return broadcastWithChannel(channelHandlerContext, message);
    }

}
